﻿using System;
using System.Collections.Generic;
using System.Disposables;
using System.Linq;

namespace SoftSource.Reactive
{
    public sealed class DeferredObservable<T> : IDeferredObservable<T>
    {
        #region Fields
        private object setGate = new object();

        private bool set = false;
        private List<Action> subscriptions = new List<Action>();
        private Func<IObserver<T>, IDisposable> subscribe = null;
        #endregion Fields

        public DeferredObservable(Func<IObserver<T>, IDisposable> subscribe)
        {
            this.subscribe = subscribe;
        }

        private IDisposable AttatchObserver(IObserver<T> observer)
        {
            MutableDisposable mutable = new MutableDisposable();

            // Here we store the actual subscription action to be performed later.
            // Note that mutable.Replace will dispose the current IDisposable
            Action attatch = () => mutable.Replace(this.subscribe(observer));

            // Don't allow the flag to be set while adding an item to the bucket.
            lock (setGate)
            {
                // Defer the subscription by adding the subcribe action to the bucket
                this.subscriptions.Add(attatch);
            }

            // If the subscriber disposes before it has been attatched, then remove from the attatch list.
            // This will get called from the attatch function because of the mutable.Replace method.
            // To prevent that, we lock and check the flag
            mutable.Replace(Disposable.Create(() =>
                {
                    lock (setGate)
                    {
                        // If the subscriber is currently being subscribed, 
                        // this disposable will be executed because of the mutable.Replace from above.
                        //
                        // When the subscritions are taking place, 
                        // we lock on a gate and set a flag.
                        if (!set)
                            this.subscriptions.Remove(attatch);
                    }
                }));
            return mutable;
        }

        private void SubscribePendingObservers()
        {
            if (this.subscriptions.Count > 0)
            {
                // Subscribe each pending observer
                this.subscriptions.Run(subscribe => subscribe());

                // Now that every one has been subscribed, clear the list of pending observers
                this.subscriptions.Clear();
            }
        }

        #region IDeferedObservable<T> Members

        public void Set()
        {
            lock (setGate) // Lock on the gate to prevent others from messing up state
            {
                set = true; // Set the flag that says we are now subscribing
                this.SubscribePendingObservers();
            }
        }

        #endregion

        #region IObservable<T> Members

        public IDisposable Subscribe(IObserver<T> observer)
        {
            lock (this.setGate)
            {
                if (this.set)
                {
                    return this.subscribe(observer);
                }
                else
                {
                    return this.AttatchObserver(observer);
                }
            }
        }

        #endregion
    }
}